package com.dingding.javabasic.线程池.手写线程池;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author liugongding
 * @description:
 * @date 2022/8/27 5:31 PM
 */
public class ThreadPoolTrader implements Executor {

    //线程池中工作线程的数量
    private final AtomicInteger ctl = new AtomicInteger(0);

    //核心线程数量
    private volatile int  corePoolSize;

    //最大线程数量
    private volatile int maximumPoolSize;

    //阻塞队列
    private final BlockingQueue<Runnable> workQueue;

    public ThreadPoolTrader(int corePoolSize, int maximumPoolSize, BlockingQueue<Runnable> workQueue) {
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
    }


    @Override
    public void execute(Runnable command) {
        //获取工作线程的数量
        int c = ctl.get();
        if (c < corePoolSize) {
            //使用核心线程执行任务
            boolean r = addWorker(command);
            if (!r) {
                reject();
            }
            return;
        }
        //将任务添加到阻塞队列（workQueue.offer()不会处于阻塞状态）
        boolean offer = workQueue.offer(command);
        if (!offer) {
            boolean r = addWorker(command);
            if (!r) {
                reject();
            }
        }
    }

    private boolean addWorker(Runnable firstTask) {
        //工作线程的数量达到最线程数时，则添加任务失败
        if (ctl.get() >= maximumPoolSize) return false;

        Worker worker = new Worker(firstTask);
        worker.thread.start();
        ctl.incrementAndGet();
        return true;
    }

    private final class Worker implements Runnable {
        final Thread thread;
        Runnable firstTask;

        public Worker(Runnable firstTask) {
            this.thread = new Thread(this);
            this.firstTask = firstTask;
        }

        @Override
        public void run() {
            Runnable task = firstTask;
            try {
                while (task != null || (task = getTask()) != null) {
                    task.run();
                    if (ctl.get() > maximumPoolSize) {
                        break;
                    }
                    task = null;
                }
            } finally {
                ctl.decrementAndGet();
            }
        }

        private Runnable getTask() {
            while (true) {
                try {
                    System.out.println("workQueue.size：" + workQueue.size());
                    return workQueue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private void reject() {
        throw new RuntimeException("Error！ctl.count：" + ctl.get() + " workQueue.size：" + workQueue.size());
    }

    public static void main(String[] args) throws Exception{
        ThreadPoolTrader threadPoolTrader = new ThreadPoolTrader
                (2, 2, new ArrayBlockingQueue<Runnable>(10));

        for (int i = 0; i < 10; i++) {
            int finalI = i;
            Runnable a = () -> {
                try {
                    Thread.sleep(1500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("任务编号：" + finalI);
            };
            threadPoolTrader.execute(a);

        }
    }
}
